package com.fyy.kafuka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

/**
 * @Author fyy
 * @Description 简单消费者
 * @Date 2020/11/6 9:23
 */
public class MyConsumer {

    public static void main(String[] args) {
        //创建消费者配置信息
        Properties config = new Properties();
        //给配置信息赋值
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"39.101.190.227:9092"); //集群地址
        config.put(ConsumerConfig.GROUP_ID_CONFIG,"Kafka_test"); //消费者组
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true); //开启消费者自动提交配置
        config.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000); //自动提交延时，单位ms
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest"); //重置消费者offset
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer"); //键反序列化
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer"); //值反序列化
        //实例化一个消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config);
        //订阅主题
        Map<TopicPartition,Long> currentOffset = new HashMap<TopicPartition,Long>();
        consumer.subscribe(Arrays.asList("aaa", "test"), new ConsumerRebalanceListener() {
            //在Rebalance之前调用
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {
                commitoffset(currentOffset);
            }
            //在Rebalance之后调用
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                currentOffset.clear(); //清空
                for (TopicPartition partition : collection) {
                    consumer.seek(partition,getOffset(partition)); //定位到最新的offset位置继续消费
                }
            }
        });
        while (true){ //阻塞程序，不断读取数据
            //获取数据
            ConsumerRecords<String, String> consumerRecords = consumer.poll(1000); //设置延时
            //解析consumerRecords数据
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println("key:" + consumerRecord.key());
                System.out.println("topic:" + consumerRecord.topic());
                System.out.println("value:" + consumerRecord.value());
                currentOffset.put(new TopicPartition(consumerRecord.topic(),
                        consumerRecord.partition()),consumerRecord.offset());
            }
            commitoffset(currentOffset); //异步提交
            //同步提交：阻塞当前线程，直到offset提交成功
            consumer.commitSync();
            //异步提交
            consumer.commitAsync(new OffsetCommitCallback() {
                @Override
                public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
                    if(e != null){
                        System.out.println("commit failed for " + offsets);
                    }
                }
            });

        }

        //关闭连接
        //consumer.close();
    }

    ///////////////自定义实现offset的存取//////////////
    private static long getOffset(TopicPartition partition){
        //从数据库/缓存中读取最新的offset
        return 0;
    }

    private static void commitoffset(Map<TopicPartition,Long> currentOffset){
        //写入自己维护的最新的offset
    }

}
